package com.sojoys.artifact.third.activemq;

import javax.jms.BytesMessage;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.jms.core.JmsTemplate;

import com.sojoys.artifact.constant.ArtifactErrorCode;
import com.sojoys.artifact.third.rpc.util.SerializationUtil;
import com.sojoys.artifact.tools.ToolError;


/**
 * @author : DengYing
 * @CreateDate : 2017年4月26日 下午4:19:03
 * @Description ：Please describe this document
 */
public class ActiveMQSender {
	/** 订阅、发布 */
	JmsTemplate topicJmsTemplate;
	/** 点对点 */
	JmsTemplate queueJmsTemplate;

	

	public void setTopicJmsTemplate(JmsTemplate topicJmsTemplate) {
		this.topicJmsTemplate = topicJmsTemplate;
	}

	public void setQueueJmsTemplate(JmsTemplate queueJmsTemplate) {
		this.queueJmsTemplate = queueJmsTemplate;
	}

	/**
	 * 发送消息
	 * @param destination 	目的地
	 * @param msg			消息体
	 * @param topic			广播
	 */
	public void convertAndSend(String destination, ActiveMQRequest msg,boolean topic) {
		if (topic) {
			topicJmsTemplate.convertAndSend(destination, msg);
		}else {
			queueJmsTemplate.convertAndSend(destination, msg);			
		}
	}

//	/**
//	 * 发送消息
//	 * @param msg
//	 * @param topic
//	 */
//	public void convertAndSend(ActiveMQRequest msg,boolean topic) {
//		if (topic) {
//			topicJmsTemplate.convertAndSend(msg);
//		}else {
//			queueJmsTemplate.convertAndSend(msg);			
//		}
//	}

	
//	public <T> T sendAndReceive(ActiveMQRequest msg) throws Throwable {
//		Message message = queueJmsTemplate.sendAndReceive((Session session) -> {
//			BytesMessage byteMsg = session.createBytesMessage();
//			byteMsg.writeBytes(SerializationUtil.serialize(msg));
//			return byteMsg;
//		});
//		return encoder(message);
//	}
	
	/**
	 * 发送&接收
	 * @param destination	目的地
	 * @param msg			消息体
	 * @return
	 * @throws Throwable
	 */
	public <T> T sendAndReceive(String destination,ActiveMQRequest msg) throws Throwable {
		Message message = queueJmsTemplate.sendAndReceive(destination,(Session session) -> {
			BytesMessage byteMsg = session.createBytesMessage();
			byteMsg.writeBytes(SerializationUtil.serialize(msg));
			return byteMsg;
		});
		
		
		return encoder(message);
	}

	@SuppressWarnings("unchecked")
	private <T> T encoder(Message message) throws Throwable{
		if (message == null) {
			throw ToolError.newLogicError(ArtifactErrorCode.SERVER_BUSY,"接收消息超时");
		}
		if (message instanceof BytesMessage) {
			BytesMessage bytesMessage = (BytesMessage) message;
			byte[] b = new byte[(int) bytesMessage.getBodyLength()];
			while (bytesMessage.readBytes(b) != -1) { 
				ActiveMQResponse response = SerializationUtil.deserialize(b, ActiveMQResponse.class);
				// 返回 RPC 响应结果
                if (response.hasException()) {
                    throw response.getException();
                } else {
                    return (T) response.getResult();
                }
			}
		}
		return null;
	}
}
